package com.dc.trans.steps.rowlevellog.filestorage;

import java.io.File;
import java.util.Collection;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.log4j.Logger;

import com.dc.trans.steps.rowlevellog.logprocess.LogSenderThreadPool;
/**
 * 
 * @author wentionlin 2012-8-9
 */
public class FileStoragePoolExecutor {
	
	protected static Logger logger = Logger.getLogger(FileStoragePoolExecutor.class);
	
	public static  FileStoragePoolExecutor executor;
	
	public static PropertiesConfiguration prop;
	
	public static String encoding = System.getProperty("file.encoding");
	
	static{
		String configPath = new File("").getAbsolutePath() + File.separator + "etc" + File.separator + "rowlevellog" + File.separator;
		prop = new PropertiesConfiguration();
		try {
			prop.load(configPath + "row_log_config.properties");
			encoding = prop.getString("logFileEncoding");
			executor = new FileStoragePoolExecutor(prop.getInt("fileStoragePoolSize"),configPath + "templogfiles",prop.getString("statusOfToBeSend"),prop.getString("logFileExtensionName"),encoding);
			
			executor.setUseZip("Y".equalsIgnoreCase(prop.getString("useZip")) ? true:false);
			executor.setRowsizeLimit(1024*1024*prop.getInt("rowsizeLimit"));
			executor.setTotalsizeLimit(1024*1024*prop.getInt("totalsizeLimit"));
			executor.setCloseTimeLimit(prop.getInt("closeTimeLimit"));
			String timeUnit = prop.getString("timeUnit");
			if("second".equalsIgnoreCase(timeUnit)){
				executor.setCloseTimeLimitUnit(TimeUnit.SECONDS);
			}else if("minute".equalsIgnoreCase(timeUnit)){
				executor.setCloseTimeLimitUnit(TimeUnit.MINUTES);
			}else if("day".equalsIgnoreCase(timeUnit)){
				executor.setCloseTimeLimitUnit(TimeUnit.DAYS);
			}else if("hour".equalsIgnoreCase(timeUnit)){
				executor.setCloseTimeLimitUnit(TimeUnit.HOURS);
			}
			executor.setDaemonThread("Y".equalsIgnoreCase(prop.getString("daemonThread")) ? true:false);
			executor.setListener(new FileStorageListener(){
				public void onDatafileCreated(File datafile) {
					if(datafile.getAbsolutePath() != null && datafile.getAbsolutePath().endsWith(".txt")){
						LogSenderThreadPool.sendLogFileAsync(datafile.getAbsolutePath());
					}
				}
			});
		} catch (ConfigurationException e) {
			e.printStackTrace();
			logger.error(e);
		}
	}
	
	public void insert(final String row){
		poolExecutor.execute(new Runnable(){
			public void run() {
				try {
					FileStorageThread fsThread = (FileStorageThread)Thread.currentThread();
					fsThread.fileStorage.insert(row);
					totalSize.getAndIncrement();
				} catch (Exception e) {
					e.printStackTrace();
					logger.error("Insert data occur error : "+e.getMessage());
				}
			}
		});
	}
	
	public void insert(final String[] row){
		poolExecutor.execute(new Runnable(){
			public void run() {
				try {
					FileStorageThread fsThread = (FileStorageThread)Thread.currentThread();
					fsThread.fileStorage.insert(row);
					totalSize.getAndIncrement();
				} catch (Exception e) {
					logger.error("Insert data occur error : "+e.getMessage());
				}
			}
		});
	}
	public void WS2APP(Runnable task) {
		poolExecutor.execute(task);
		
	}
	public FileStoragePoolExecutor(int maxStorageSize,String storeDir,String filePrefix, String fileSuffix, String encoding) {
		setPoolSize(maxStorageSize);
		setStoreDirectory(storeDir);
		setFilePrefix(filePrefix);
		setFileSuffix(fileSuffix);
		FileStoragePoolExecutor.encoding = encoding;
		poolExecutor = new ThreadPoolExecutor(poolSize, poolSize,getCloseTimeLimit(),getCloseTimeLimitUnit(), new OfferIsBlockingQueue(getConcurrentCacheSize()), new ThreadFactory() {
			public Thread newThread(Runnable r) {
				return new FileStorageThread(r);
			}
		});//,new ThreadPoolExecutor.CallerRunsPolicy()
		
		poolExecutor.allowCoreThreadTimeOut(true);
	}

	public void shutdown(){
		poolExecutor.shutdownNow();
		System.gc();
	}
	
	private class DefaultFileStorageConfiger implements FileStorageConfiger{
		@Override
		public FileStorage config(FileStorage fileStorage) {
			fileStorage.setUseBuffer(isUseBuffer());
			fileStorage.setUseZip(isUseZip());
			fileStorage.setColmSeparator(getColmSeparator());
			fileStorage.setRowsizeLimit(getRowsizeLimit());
			fileStorage.setTotalsizeLimit(getTotalsizeLimit());
			fileStorage.setCloseDatafileTime(getCloseTimeLimit(), getCloseTimeLimitUnit());
			fileStorage.setListener(getListener());
			return fileStorage;
		};
	}
	
	private static class OfferIsBlockingQueue extends ArrayBlockingQueue<Runnable>{

		private static final long serialVersionUID = 1L;

		public OfferIsBlockingQueue(int capacity, boolean fair, Collection<? extends Runnable> c) {
			super(capacity, fair, c);
		}

		public OfferIsBlockingQueue(int capacity, boolean fair) {
			super(capacity, fair);
		}

		public OfferIsBlockingQueue(int capacity) {
			super(capacity);
			logger.info("FileStoragePoolExecutor Queue message:"+capacity);//add by hsg
		}
		
		@Override
		public boolean offer(Runnable task) {
			try {
				put(task);
			} catch (Exception e) {
				logger.info("FileStoragePoolExecutor error message:"+e);//add by hsg
				return false;
			}
			return true;
		}
	}
	
	private class FileStorageThread extends Thread{
		FileStorage fileStorage = null;
		public FileStorageThread(Runnable target) {
			super(target);
			setDaemon(isDaemonThread());
			fileStorage = getFileStorageConfiger().config(new FileStorage(getStoreDirectory(),getFilePrefix(),getFileSuffix(),encoding));
		}
		protected void finalize() throws Throwable {
			super.finalize();
			try {
				fileStorage.stopStorage();
			} catch (Exception e) {
				logger.error("Finalize file storage thread occur error : "+e.getMessage());
			}
		}
	}
	

	private AtomicLong totalSize = new AtomicLong(0);

	private int poolSize = 5;

	private int concurrentCacheSize = 100*10000;
	
	private ThreadPoolExecutor poolExecutor = null;

	private FileStorageConfiger fileStorageConfiger = null;
	
	private String storeDirectory = null;
	
	private String filePrefix = "filestore";

	private String fileSuffix = ".fsd";
	
	private FileStorageListener listener = null;
	
	private String colmSeparator = ",";
	
	private boolean useBuffer = true; 

	private boolean useZip = true;
	
	private long closeTimeLimit = 5;
	
	private TimeUnit closeTimeLimitUnit = TimeUnit.MINUTES;
	
	private long totalsizeLimit = 1024*20;

	private long rowsizeLimit = 1024*1024;
	
	private boolean daemonThread = true;
	
	public boolean isDaemonThread() {
		return daemonThread;
	}

	public void setDaemonThread(boolean daemonThread) {
		this.daemonThread = daemonThread;
	}

	public int getCurrentPoolSize(){
		return poolExecutor.getPoolSize();
	}
	
	public int getActiveCount(){
		return poolExecutor.getActiveCount();
	}
	
	public int getConcurrentCacheSize() {
		return concurrentCacheSize;
	}

	public void setConcurrentCacheSize(int concurrentCacheSize) {
		this.concurrentCacheSize = concurrentCacheSize;
	}

	public long getTotalSize(){
		return totalSize.get();
	}
	
	public int getCurrentTaskSize(){
		return poolExecutor.getQueue().size();
	}
	
	public int getPoolSize() {
		return poolSize;
	}

	public void setPoolSize(int poolSize) {
		this.poolSize = poolSize;
	}


	public FileStorageConfiger getFileStorageConfiger() {
		if(fileStorageConfiger==null){
			fileStorageConfiger = new DefaultFileStorageConfiger();
		}
		return fileStorageConfiger;
	}


	public void setFileStorageConfiger(FileStorageConfiger fileStorageConfiger) {
		this.fileStorageConfiger = fileStorageConfiger;
	}


	public String getStoreDirectory() {
		return storeDirectory;
	}


	public void setStoreDirectory(String storeDirectory) {
		this.storeDirectory = storeDirectory;
	}


	public String getFilePrefix() {
		return filePrefix;
	}


	public void setFilePrefix(String filePrefix) {
		this.filePrefix = filePrefix;
	}


	public String getFileSuffix() {
		return fileSuffix;
	}


	public void setFileSuffix(String fileSuffix) {
		this.fileSuffix = fileSuffix;
	}


	public FileStorageListener getListener() {
		return listener;
	}


	public void setListener(FileStorageListener listener) {
		this.listener = listener;
	}


	public String getColmSeparator() {
		return colmSeparator;
	}


	public void setColmSeparator(String colmSeparator) {
		this.colmSeparator = colmSeparator;
	}


	public boolean isUseBuffer() {
		return useBuffer;
	}


	public void setUseBuffer(boolean useBuffer) {
		this.useBuffer = useBuffer;
	}


	public boolean isUseZip() {
		return useZip;
	}


	public void setUseZip(boolean useZip) {
		this.useZip = useZip;
	}


	public long getCloseTimeLimit() {
		return closeTimeLimit;
	}


	public void setCloseTimeLimit(long closeTimeLimit) {
		this.closeTimeLimit = closeTimeLimit;
	}


	public TimeUnit getCloseTimeLimitUnit() {
		return closeTimeLimitUnit;
	}


	public void setCloseTimeLimitUnit(TimeUnit closeTimeLimitUnit) {
		this.closeTimeLimitUnit = closeTimeLimitUnit;
	}


	public long getTotalsizeLimit() {
		return totalsizeLimit;
	}


	public void setTotalsizeLimit(long totalsizeLimit) {
		this.totalsizeLimit = totalsizeLimit;
	}


	public long getRowsizeLimit() {
		return rowsizeLimit;
	}


	public void setRowsizeLimit(long rowsizeLimit) {
		this.rowsizeLimit = rowsizeLimit;
	}


	public void setCloseDatafileTime(long closeTime,TimeUnit unit){
		closeTimeLimit = closeTime;
		closeTimeLimitUnit = unit;
	}

	public long getClostTimeLimitMilliseconds(){
		return TimeUnit.MILLISECONDS.convert(closeTimeLimit, closeTimeLimitUnit);
	}

	
}
